package demo;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.CountDownLatch2;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * 异步消息
 * 使用异步线程发送消息，send方法只会将Message放入异步线程中
 * 放完及执行完成了，所以主线程末尾的等待是必要的
 */
public class AsyncProducerMessage {
    public static void main(String[] args) throws Exception {

        DefaultMQProducer mqProducer = new DefaultMQProducer("AsyncGroup");
        mqProducer.setNamesrvAddr("192.168.86.104:10001");
        mqProducer.start();
        int msgCount = 100;
        CountDownLatch2 countDownLatch2 = new CountDownLatch2(msgCount);
        for(int i=0;i<msgCount;i++) {
            final int index = i;
            Message msg = new Message("demo_topic","tag1",
                    ("Async_HelloWorld"+i).getBytes(StandardCharsets.UTF_8));
            //SendCallback接口有两个方法分别是在发送成功和异常调用，
            //这里虽然是循环，但是异步不一定会按照循环的顺序发送
            //线程名称是NettyClient和group没有关系
            mqProducer.send(msg,new SendCallback(){
                @Override
                public void onSuccess(SendResult sendResult) {
                    countDownLatch2.countDown();
                    String threadName = Thread.currentThread().getName();
                    System.out.printf("%-10d OK %s %s %n",index,threadName,sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable e) {
                    e.printStackTrace();
                    countDownLatch2.countDown();
                    String threadName = Thread.currentThread().getName();
                    System.out.printf("%-10d ERROR %s %s %n", index,threadName,e.getMessage());
                }
            });

        }
        //注意这里例子是异步，send方法不需要等待Broker返回就执行完了，主线程需要等待异步线程发送消息给MQ
        //才能mqProducer被关闭
        countDownLatch2.await(10, TimeUnit.SECONDS);
        mqProducer.shutdown();
    }
}
